/*
 * Copyright (C) 2010-2101 Alibaba Group Holding Limited.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.alibaba.otter.manager.biz.config.canal.impl;

import com.alibaba.otter.canal.instance.manager.model.Canal;
import com.alibaba.otter.canal.instance.manager.model.CanalParameter;
import com.alibaba.otter.manager.biz.common.exceptions.ManagerException;
import com.alibaba.otter.manager.biz.common.exceptions.RepeatConfigureException;
import com.alibaba.otter.manager.biz.config.autokeeper.AutoKeeperClusterService;
import com.alibaba.otter.manager.biz.dto.CommonSearchDTO;
import com.alibaba.otter.manager.biz.config.canal.CanalService;
import com.alibaba.otter.manager.biz.config.canal.dal.CanalDao;
import com.alibaba.otter.manager.biz.config.canal.dal.dataobject.CanalDO;
import com.alibaba.otter.manager.biz.config.node.impl.NodeServiceImpl;
import com.alibaba.otter.shared.arbitrate.ArbitrateViewService;
import com.alibaba.otter.shared.common.model.autokeeper.AutoKeeperCluster;
import com.alibaba.otter.shared.common.utils.Assert;
import com.alibaba.otter.shared.common.utils.JsonUtils;
import com.mybatisflex.core.paginate.Page;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.domain.Pageable;
import org.springframework.stereotype.Service;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.TransactionCallbackWithoutResult;
import org.springframework.transaction.support.TransactionTemplate;

import java.sql.Date;
import java.util.*;

/**
 * @author sarah.lij 2012-7-25 下午04:04:43
 */
@Service
public class CanalServiceImpl implements CanalService {

    private static final Logger logger = LoggerFactory.getLogger(NodeServiceImpl.class);

    private final CanalDao canalDao;


    private final TransactionTemplate transactionTemplate;

    private final AutoKeeperClusterService autoKeeperClusterService;

    private final ArbitrateViewService arbitrateViewService;

    public CanalServiceImpl(CanalDao canalDao,
                            TransactionTemplate transactionTemplate,
                            AutoKeeperClusterService autoKeeperClusterService,
                            ArbitrateViewService arbitrateViewService) {
        this.canalDao = canalDao;
        this.transactionTemplate = transactionTemplate;
        this.autoKeeperClusterService = autoKeeperClusterService;
        this.arbitrateViewService = arbitrateViewService;
    }

    /**
     * 添加
     */
    @Transactional(rollbackFor = Exception.class)
    public void create(final Canal canal) {
        try {
            CanalDO canalDO = modelToDo(canal);
            canalDO.setId(0L);
            if (!canalDao.checkUnique(canalDO)) {
                String exceptionCause = "exist the same repeat canal in the database.";
                logger.warn("WARN ## " + exceptionCause);
                throw new RepeatConfigureException(exceptionCause);
            }
            canalDao.uniqueInsert(canalDO);
            canal.setId(canalDO.getId());
        } catch (RepeatConfigureException rce) {
            throw rce;
        } catch (Exception e) {
            logger.error("ERROR ## create canal has an exception!");
            throw new ManagerException(e);
        }

    }

    /**
     * 删除
     */
    public void remove(final Long canalId) {
        Assert.assertNotNull(canalId);
        transactionTemplate.execute(new TransactionCallbackWithoutResult() {

            protected void doInTransactionWithoutResult(TransactionStatus status) {

                try {
                    Canal canal = findById(canalId);
                    canalDao.deleteById(canalId);
                    arbitrateViewService.removeCanal(canal.getName()); // 删除canal节点信息
                } catch (Exception e) {
                    logger.error("ERROR ## remove canal(" + canalId + ") has an exception!");
                    throw new ManagerException(e);
                }
            }
        });

    }

    /**
     * 修改
     */
    @Transactional
    public void modify(final Canal canal) {
        Assert.assertNotNull(canal);
        try {
            CanalDO canalDo = modelToDo(canal);
            if (canalDao.checkUnique(canalDo)) {
                canalDao.insertOrUpdate(canalDo,true);
            } else {
                String exceptionCause = "exist the same repeat canal in the database.";
                logger.warn("WARN ## " + exceptionCause);
                throw new RepeatConfigureException(exceptionCause);
            }
        } catch (RepeatConfigureException rce) {
            throw rce;
        } catch (Exception e) {
            logger.error("ERROR ## modify canal(" + canal.getId() + ") has an exception!");
            throw new ManagerException(e);
        }
    }

    public List<Canal> listByIds(Long... identities) {

        List<Canal> canals = new ArrayList<>();
        try {
            List<CanalDO> canalDos;
            if (identities.length < 1) {
                canalDos = canalDao.selectAll();
                if (canalDos.isEmpty()) {
                    logger.debug("DEBUG ## couldn't query any canal, maybe hasn't create any canal.");
                    return canals;
                }
            } else {
                canalDos = canalDao.selectListByIds(Arrays.asList(identities));
                if (canalDos.isEmpty()) {
                    String exceptionCause = "couldn't query any canal by canalIds:" + Arrays.toString(identities);
                    logger.error("ERROR ## " + exceptionCause);
                    throw new ManagerException(exceptionCause);
                }
            }
            canals = doToModel(canalDos);
        } catch (Exception e) {
            logger.error("ERROR ## query channels has an exception!");
            throw new ManagerException(e);
        }

        return canals;
    }

    public List<Canal> listAll() {
        return listByIds();
    }

    public Canal findById(Long canalId) {
        Assert.assertNotNull(canalId);
        List<Canal> canals = listByIds(canalId);
        if (canals.size() != 1) {
            String exceptionCause = "query canalId:" + canalId + " return null.";
            logger.error("ERROR ## {}", exceptionCause);
            throw new ManagerException(exceptionCause);
        }

        return canals.get(0);
    }

    public Canal findByName(String name) {
        Assert.assertNotNull(name);
        CanalDO canalDo = canalDao.selectByName(name);
        if (canalDo == null) {
            String exceptionCause = "query name:" + name + " return null.";
            logger.error("ERROR ## {}", exceptionCause);
            throw new ManagerException(exceptionCause);
        }

        return doToModel(canalDo);
    }


    @Override
    public List<Canal> listByCondition(Map<String, Object> condition) {
        List<CanalDO> canalDos = canalDao.selectListByMap(condition);
        if (canalDos.isEmpty()) {
            logger.debug("DEBUG ## couldn't query any canal by the condition:{}", JsonUtils.marshalToString(condition));
            return new ArrayList<Canal>();
        }

        return doToModel(canalDos);
    }

    @Override
    public Page<Canal> canalPage(CommonSearchDTO searchVo, Pageable pageable) {
        Page<CanalDO> pagedList = canalDao.canalPage(searchVo, pageable);
        return pagedList.map(this::doToModel);
    }

    /**
     * 用于Model对象转化为DO对象
     *
     * @param canal
     * @return CanalDO
     */
    private CanalDO modelToDo(Canal canal) {
        CanalDO canalDo = new CanalDO();
        try {
            canalDo.setId(canal.getId());
            canalDo.setName(canal.getName());
            canalDo.setDescription(canal.getDesc());
            canalDo.setParameters(canal.getCanalParameter());
        } catch (Exception e) {
            logger.error("ERROR ## change the canal Model to Do has an exception");
            throw new ManagerException(e);
        }
        return canalDo;
    }

    /**
     * 用于DO对象转化为Model对象
     *
     * @param canalDo
     * @return Canal
     */
    private Canal doToModel(CanalDO canalDo) {
        Canal canal = new Canal();
        try {
            canal.setId(canalDo.getId());
            canal.setName(canalDo.getName());
            canal.setDesc(canalDo.getDescription());
            canal.setStatus(canalDo.getStatus());
            CanalParameter parameter = canalDo.getParameters();
            AutoKeeperCluster zkCluster = autoKeeperClusterService.findAutoKeeperClusterById(parameter.getZkClusterId());
            if (zkCluster != null) {
                parameter.setZkClusters(Collections.singletonList(StringUtils.join(zkCluster.getServerList(), ',')));
            }
            canal.setCanalParameter(canalDo.getParameters());
            canal.setGmtCreate(Date.from(canalDo.getGmtCreate()));
            canal.setGmtModified(Date.from(canalDo.getGmtModified()));
        } catch (Exception e) {
            logger.error("ERROR ## change the canal Do to Model has an exception");
            throw new ManagerException(e);
        }

        return canal;
    }

    private List<Canal> doToModel(List<CanalDO> canalDos) {
        List<Canal> canals = new ArrayList<>();
        for (CanalDO canalDo : canalDos) {
            canals.add(doToModel(canalDo));
        }
        return canals;
    }

}
